package com.pk.flink.apps.consumerGroup.consumerMoreThanPartitions;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;


public class KafkaConsumerGroupProducerApp {

    private static String SERVERS = "master:9092,slave1:9092,slave2:9092";
    private static String TOPIC = "pk-3-1";
    KafkaProducer<String, String> producer = null;

    public static void main(String[] args){
        new KafkaConsumerGroupProducerApp().run();
    }

    private void run(){
        Properties props = new Properties();
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVERS);
        props.put(ProducerConfig.ACKS_CONFIG, "1");
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);

        producer = new KafkaProducer<>(props);


        try {

            for(int i=0;i<10;i++)
            {

                String key = "key" + i;
                String value = "value" + i;

                producer.send(new ProducerRecord<String,String>(TOPIC, key,value));
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        } catch (KafkaException e) {
            // We can't recover from these exceptions, so our only option is to close the producer and exit.
            System.err.println(e.getMessage());
        }// For all other exceptions, just abort the transaction and try again.
        finally {
            producer.close();
        }

    }

}
